package com.ruoyi.iot.service.impl;

import cn.hutool.json.JSONObject;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

/**
 * eKuiper 规则接口
 *
 * @author Howe
 * @since 2025/7/17
 */
@Service
public class IotEKuiperService {

    private final RestTemplate restTemplate = new RestTemplate();
    /**
     * eKuiper API 地址
     */
    private final String ekuiperUrl = "http://47.120.66.54:9081";

    /**
     * 创建 eKuiper 规则
     *
     * @param ruleName 规则名称
     * @param sql      SQL 语句
     * @param actions  actions对象数组（List/JSONArray）
     * @return eKuiper返回结果
     */
    public String createRule(String ruleName, String sql, Object actions) {
        String url = ekuiperUrl + "/rules";
        cn.hutool.json.JSONObject ruleJson = new cn.hutool.json.JSONObject();
        ruleJson.put("id", ruleName);
        ruleJson.put("sql", sql);
        ruleJson.put("actions", actions);
        return restTemplate.postForObject(url, ruleJson, String.class);
    }

    /**
     * 删除 eKuiper 规则
     */
    public void deleteRule(String ruleName) {
        String url = ekuiperUrl + "/rules/" + ruleName;
        restTemplate.delete(url);
    }

    /**
     * 查询 eKuiper 规则
     */
    public String getRule(String ruleName) {
        String url = ekuiperUrl + "/rules/" + ruleName;
        return restTemplate.getForObject(url, String.class);
    }

    /**
     * 查询 eKuiper 中所有规则
     *
     * @return 所有规则的 JSON 字符串
     * @author Howe
     * @since 2025/7/17
     */
    public String getAllRules() {
        String url = ekuiperUrl + "/rules";
        return restTemplate.getForObject(url, String.class);
    }

    /**
     * 自动检测并创建 eKuiper stream（如不存在）
     *
     * @param streamName 流名称
     * @param fields     字段定义（如 "field1 FLOAT, field2 STRING"）
     * @param datasource MQTT topic
     * @return 创建结果
     */
    public String createStreamIfNotExists(String streamName, String fields, String datasource) {
        String getUrl = ekuiperUrl + "/streams/" + streamName;
        try {
            String getResult = restTemplate.getForObject(getUrl, String.class);
            if (getResult != null && getResult.contains(streamName)) {
                return "stream already exists";
            }
        } catch (org.springframework.web.client.HttpClientErrorException e) {
            // 流不存在时 eKuiper 返回 400 错误，这是正常情况，继续创建流
            if (e.getStatusCode().value() == 400 && e.getResponseBodyAsString().contains("is not found")) {
                // 流不存在，继续创建
            } else {
                // 其他错误则重新抛出
                throw e;
            }
        }
        // 构造 CREATE STREAM SQL
        String sql = String.format("CREATE STREAM %s (%s) WITH (DATASOURCE=\"%s\", FORMAT=\"json\")", streamName,
                fields, datasource);
        JSONObject body = new JSONObject();
        body.put("sql", sql);
        String postUrl = ekuiperUrl + "/streams";
        return restTemplate.postForObject(postUrl, body, String.class);
    }

    /**
     * 自动检测并删除 eKuiper stream（如存在）
     *
     * @param streamName 流名称
     * @return 删除结果
     */
    public String deleteStreamIfExists(String streamName) {
        String getUrl = ekuiperUrl + "/streams/" + streamName;
        try {
            String getResult = restTemplate.getForObject(getUrl, String.class);
            if (getResult != null && getResult.contains(streamName)) {
                String delUrl = ekuiperUrl + "/streams/" + streamName;
                restTemplate.delete(delUrl);
                return "stream deleted";
            }
        } catch (org.springframework.web.client.HttpClientErrorException e) {
            // 流不存在时 eKuiper 返回 400 错误，这是正常情况
            if (e.getStatusCode().value() == 400 && e.getResponseBodyAsString().contains("is not found")) {
                return "stream not exists";
            } else {
                // 其他错误则重新抛出
                throw e;
            }
        }
        return "stream not exists";
    }

    /**
     * 创建 eKuiper 规则（自动创建 stream）
     *
     * @param ruleName     规则名称
     * @param sql          SQL 语句
     * @param actions      actions对象数组（List/JSONArray）
     * @param streamFields 流字段定义（如 "field1 FLOAT, field2 STRING"）
     * @param datasource   MQTT topic
     * @return eKuiper返回结果
     */
    public String createRule(String ruleName, String sql, Object actions, String streamFields, String datasource) {
        // 自动创建 stream
        String streamName = getStreamNameFromSql(sql);
        createStreamIfNotExists(streamName, streamFields, datasource);
        String url = ekuiperUrl + "/rules";
        cn.hutool.json.JSONObject ruleJson = new cn.hutool.json.JSONObject();
        ruleJson.put("id", ruleName);
        ruleJson.put("sql", sql);
        ruleJson.put("actions", actions);
        return restTemplate.postForObject(url, ruleJson, String.class);
    }

    /**
     * 删除 eKuiper 规则（自动删除 stream）
     *
     * @param ruleName   规则名称
     * @param streamName 流名称
     */
    public void deleteRule(String ruleName, String streamName) {
        String url = ekuiperUrl + "/rules/" + ruleName;
        restTemplate.delete(url);
        // 自动删除 stream
        deleteStreamIfExists(streamName);
    }

    /**
     * 从 SQL 语句中提取 stream 名称（假设 FROM 后第一个单词为 stream）
     */
    private String getStreamNameFromSql(String sql) {
        if (sql == null)
            return null;
        String upper = sql.toUpperCase();
        int fromIdx = upper.indexOf("FROM");
        if (fromIdx == -1)
            return null;
        String afterFrom = sql.substring(fromIdx + 4).trim();
        String[] parts = afterFrom.split("\\s+");
        return parts.length > 0 ? parts[0] : null;
    }

}
